#pragma once
#include "server.h"
#include "Log.h"
#include "timewrap.h"

class server_proxy 
{
	int _sock;
	address _addr;
	timee _disconnect_time;
public:
	static int io_read_timeout; // ms
	static int io_write_timeout;
	server_proxy() :_sock(-1) {};
	server_proxy(const address &addr) : _sock(-1), _addr(addr) {}
	server_proxy(const server_proxy &rhs) : _sock(-1), _addr(rhs._addr) {}
	server_proxy &operator=(const server_proxy &rhs) { _sock = -1; _addr = rhs._addr; return *this; }
	~server_proxy() { if (_sock != -1) close(_sock); }
	int get_sock()
	{
		if (_sock == -1)
			reconnect();
		return _sock;
	}
	address &get_address() { return _addr; }
	
	bool check()
	{
		if (get_sock() == -1)
			return false;
		if (check_alive(get_sock()) == 1)
			return true;
		LOG_INFO_VA("server<%s,%d> down, reset it", _addr.ip().c_str(), _addr.port());
		disconnect();
		return (reconnect() == 0);
	}
	void disconnect()
	{
		if (_sock != -1) 
			close(_sock);
		_sock = -1;
		_disconnect_time = timee::now();
	}
	int reconnect()
	{
		if (timee::now() - _disconnect_time < 86000000)
			return -1;
		int fd = socket(PF_INET, SOCK_STREAM, 0);
		if (fd < 0) {
			LOG_ERROR_VA("create socket error");
			return -1;
		}
		int flag = 1;
		setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &flag, sizeof(int));
		//if (connect(fd, &_addr, _addr.length()) < 0) {
		if (connect(fd, &_addr, _addr.length()) < 0) {
			_disconnect_time = timee::now();
			LOG_ERROR_VA("connect to <%s,%d> error ", _addr.ip().c_str(), _addr.port());
			close(fd);
			return -1;
		}

		_sock = fd;
		return 0;

	}
	int reset()
	{
		disconnect();
		return reconnect();
	}
	int request(message &req_msg, response_message &res_msg)
	{
		// send_msg(msg);
		// read head, buf
		// res_msg.unmarshal(head, buf);
		if (!check()) {
			LOG_ERROR_VA("request server <%s,%d> down", _addr.ip().c_str(), _addr.port());
			return -1;
		}
		char err[128];
		if (send_msg(req_msg) <= 0) {
			res_msg.set_status(-1);
			snprintf(err, 127, "write request %d failed", req_msg.get_head().command);
			res_msg.set_error_str(err);
			disconnect();
			return -1;
		}
		LOG_INFO_VA("request msg %d wait for response", req_msg.get_id());
		int ret = recv_msg(res_msg);
		if (ret <= 0) {
			disconnect();
			return -1;
		}
		return 0;
	}
	int block_request(message &req_msg, response_message &res_msg)
	{
		// send_msg(msg);
		// read head, buf
		// res_msg.unmarshal(head, buf);
		if (!check()) {
			LOG_ERROR_VA("request server <%s,%d> down", _addr.ip().c_str(), _addr.port());
			return -1;
		}
		char err[128];
		if (send_msg(req_msg) <= 0) {
			res_msg.set_status(-1);
			snprintf(err, 127, "write request %d failed", req_msg.get_head().command);
			res_msg.set_error_str(err);
			disconnect();
			return -1;
		}
		LOG_DEBUG_VA("request msg %d wait for response", req_msg.get_id());
		int ret = recv_msg(res_msg);
		if (ret <= 0) {
			disconnect();
			return -1;
		}
		return 0;
	}

	int notify(message &msg)
	{
		return send_msg(msg);
	}
	int send_msg(message &msg)
	{
		LOG_INFO_VA("send msg %d to <%s,%d>", msg.get_id(), _addr.ip().c_str(), _addr.port());
		return ::send_msg(get_sock(), msg, io_write_timeout);
	}
	int recv_msg(message &msg)
	{
		//read head;
		//read buf of head.length size
		//msg.unmarshal(head, buf);
		return ::recv_msg(get_sock(), msg, io_read_timeout);
	}
	};

class server_proxy_duplication
{
	vector<server_proxy> _servers;
	unsigned int _turn; 
public:
	server_proxy_duplication() : _turn(0) {}
	void add_server_proxy(const server_proxy &svr) { _servers.push_back(svr); }
	server_proxy *choose_one() { return choose_policy();}
	server_proxy &operator[](int index) { return _servers[index]; }
	server_proxy *choose_policy()
	{
		int size = _servers.size();
		if (size == 0) {
			return 0;
		}
		
		for (int i = 0; i != size; i++) {
			server_proxy *svr = &(_servers[_turn++ % size]);
			if (!svr->check()) {
				LOG_WARN_VA("server proxy <%s,%d> turned down, reconnect", svr->get_address().ip().c_str(), svr->get_address().port());
				svr->reconnect();
			} else {
				return svr;
			}
		}
		return 0;
	}
};

class server_proxy_group
{
	vector<server_proxy> _servers;
public:
	void add_server_proxy(const server_proxy &svr) { _servers.push_back(svr); }
	int group_request(message &req_msg, message_builder_base &msg_builder, vector<response_message *> &results);
	int get_group_size() { return _servers.size(); }
	server_proxy &operator[](int index) { return _servers[index]; }
};

